package de.ruedigermoeller.fastcast.service.fcdemon;

import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import de.ruedigermoeller.fastcast.remoting.*;
import de.ruedigermoeller.fastcast.util.FCLog;
import de.ruedigermoeller.fastcast.util.FCUtils;

import java.io.*;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * Created with IntelliJ IDEA.
 * User: ruedi
 * Date: 9/5/13
 * Time: 7:53 PM
 * To change this template use File | Settings | File Templates.
 */
public class FCDemon extends FCTopicService {

    @Parameter(names = "-d", description = "synced directory (default: ./shared)")
    String outDir = "./shared/";

    ArrayList<ProcessEntry> processes = new ArrayList<>();
    static String args[] = {};

    public static class ProcessEntry implements Serializable {
        String name;
        int weight;
        String pid;
        transient Process proc;

        public ProcessEntry(String name, String pid, int weight, Process proc) {
            this.name = name;
            this.pid = pid;
            this.weight = weight;
            this.proc = proc;
        }

        public Process getProc() {
            return proc;
        }

        public void setProc(Process proc) {
            this.proc = proc;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public String getPid() {
            return pid;
        }

        public void setPid(String pid) {
            this.pid = pid;
        }

        public int getWeight() {
            return weight;
        }

        public void setWeight(int weight) {
            this.weight = weight;
        }

        @Override
        public String toString() {
            return "ProcessEntry{" +
                    "name='" + name + '\'' +
                    ", weight=" + weight +
                    ", pid='" + pid + '\'' +
                    '}';
        }
    }

    public static class NodeInfo implements Serializable {
        String host;
        List<ProcessEntry> processes;
        String nodeId;

        public NodeInfo(String host, List<ProcessEntry> processes, String nodeId) {
            this.host = host;
            this.processes = processes;
            this.nodeId = nodeId;
        }

        public String getNodeId() {
            return nodeId;
        }

        public String getHost() {
            return host;
        }

        public List<ProcessEntry> getProcesses() {
            return processes;
        }

        @Override
        public String toString() {
            return "NodeInfo{" +
                    "host='" + host + '\'' +
                    ", nodeId='" + nodeId + '\'' +
                    ", num processes=" + processes.size() +
                    '}';
        }
    }

    NodeInfo info;

    @Override
    public void init() {
        JCommander com = new JCommander(this,args);
        new File(outDir).mkdirs();
        String hostName = null;
        try {
            hostName = InetAddress.getLocalHost().getHostName();
            info = new NodeInfo(hostName, processes,getNodeId());
        } catch (UnknownHostException e) {
            FCLog.log(e);
            info = new NodeInfo("unknown", processes,getNodeId());
        }
    }

    @RemoteMethod(7)
    public void getProcessList(FCFutureResultHandler procList /*NodeInfo*/) {
        checkDead();
        procList.sendResult(info);
    }

    private void checkDead() {
        for (int i = 0; i < processes.size(); i++) {
            ProcessEntry processEntry = processes.get(i);
            try {
                processEntry.getProc().exitValue();
                processes.remove(i);
                i--;
            } catch (Throwable th) {
                // expected
            }
        }
    }

    @RemoteMethod(8)
    public void killProc(String id, FCFutureResultHandler string) {
        for (int i = 0; i < processes.size(); i++) {
            ProcessEntry processEntry = processes.get(i);
            if ( processEntry.getPid().equals(id) ) {
                processEntry.getProc().destroy();
                string.sendResult("Process "+id+" destroyed.");
                processes.remove(i);
                return;
            }
        }
        // if not found, don't send response
    }

    @RemoteMethod(9)
    public void reqRespTest(long nanosIn, FCFutureResultHandler res) {
        res.sendResult(nanosIn);
    }

    int count =0;
    long lastMeasure = 0;
    @RemoteMethod(10)
    public void maxMethodsCallSpeedReceiver(boolean reset) {
        if ( reset ) {
            count = 0;
            lastMeasure = System.currentTimeMillis();
        }
        count++;
        if ( count % 1000 == 0 ) {
            long now = System.currentTimeMillis();
            long diff = now - lastMeasure;
            if ( diff > 1000 ) {
                System.out.println("received "+count+" remote calls in "+diff+" ms");
                count = 0; lastMeasure=now;
            }
        }
    }

    @RemoteMethod(6)
    public void getLoad(FCFutureResultHandler intLoad) {
        int res = 0;
        for (int i = 0; i < processes.size(); i++) {
            ProcessEntry processEntry = processes.get(i);
            res += processEntry.getWeight();
        }
        intLoad.sendResult(res);
    }

    public static String findCurrentJavaBin() {
        String bootcp = System.getProperty("sun.boot.class.path");
        String[] elems = null;
        elems = bootcp.split(File.pathSeparator);
        for (int i = 0; i < elems.length; i++) {
            String elem = elems[i];
            if ( elem.indexOf("rt.jar") >= 0 ) {
                return new File(elem).getParentFile().getParentFile().getAbsolutePath()+File.separator+"bin"+File.separator+"java";
            }
        }
        return System.getenv("JAVA_HOME")+File.separator+"bin"+"java";
    }

    static SimpleDateFormat format = new SimpleDateFormat("MM/dd HH:mm:ss:SSS");

    private static class StreamGobbler extends Thread {
        InputStream is;
        String type;

        private StreamGobbler(InputStream is, String type) {
            this.is = is;
            this.type = type;
        }

        @Override
        public void run() {
            try {
                InputStreamReader isr = new InputStreamReader(is);
                BufferedReader br = new BufferedReader(isr);
                String line = null;
                while ((line = br.readLine()) != null)
                    System.out.println(type + " ["+ format.format(System.currentTimeMillis())+"] >" + line +"<");
            }
            catch (IOException ioe) {
                ioe.printStackTrace();
            }
        }
    }

    String [] splitCmdLine(String line) {
        String[] split = line.split(" ");
        ArrayList<String> res = new ArrayList<>();
        for (int i = 0; i < split.length; i++) {
            String s = split[i];
            if ( s.startsWith("\"") ) {
                s = s.substring(1);
                if ( s.endsWith("\"")) {
                    s = s.substring(0,s.length()-1);
                } else {
                    i++;
                    s = s+" "+split[i].substring(0,split[i].length()-1);
                }
            }
            if ( s.trim().length() > 0 ) {
                res.add(s);
            }
        }
        String resLine[] = new String[res.size()];
        for (int i = 0; i < res.size(); i++) {
            resLine[i] = res.get(i);
        }
        return resLine;
    }

    @RemoteMethod(11)
    public void startJavaClass(String id, String vmoptions, String className, String args, String cp, FCFutureResultHandler processEntryOrException) {
        String commandLine = findCurrentJavaBin();
        if ( FCUtils.isWindows()) {
            commandLine="\""+commandLine+".exe\"";
        }
        File base = new File(".").getAbsoluteFile();
        cp = expandCP(cp,base.getAbsolutePath());
        commandLine+=" "+vmoptions+" -cp "+cp+" "+className+" "+args;
        id= getNodeId()+"-"+id+"-"+procCount++;
        ProcessBuilder proc = new ProcessBuilder(splitCmdLine(commandLine));
        proc.directory( null );
        proc.redirectErrorStream(true);
        System.out.println("starting "+commandLine);
        try {
            Process p = proc.start();
            StreamGobbler out = new StreamGobbler(p.getInputStream(), id);
            out.start();
            ProcessEntry en = new ProcessEntry(commandLine,id,1,p);
            processes.add(en);
            processEntryOrException.sendResult(en);
        } catch (IOException e) {
            FCLog.log(e);
            processEntryOrException.sendResult(e.getClass().getSimpleName()+"#"+e.getMessage());
        }
    }

    private String expandCP(String cp, String base) {
        if ( FCUtils.isWindows()) {
            cp = cp.replace("/", File.separator);
            cp = cp.replace(":", ";");
        } else {
            cp = cp.replace("\\", "/");
            cp = cp.replace(";", ":");
        }
        String path[] = cp.split(File.pathSeparator);
        ArrayList<String> res = new ArrayList<>();
        for (int i = 0; i < path.length; i++) {
            String s = path[i];
            if ( s.endsWith("*") ) {
                String par = s.substring(0,s.length() - 1);
                String[] list = new File(par).list();
                for (int j = 0; list != null && j < list.length; j++) {
                    String jar = list[j];
                    if ( jar.toLowerCase().endsWith(".jar") ) {
                        res.add(par+jar);
                    }
                }
            } else {
                res.add(s);
            }
        }
        String newCP = "";
        for (int i = 0; i < res.size(); i++) {
            Object o = res.get(i);
            newCP += o+File.pathSeparator;
        }
        return newCP;
    }

    int procCount = 0;
    @RemoteMethod(5)
    public void startProcess(int weight, String id, String commandLine, FCFutureResultHandler processEntryOrException) {
        id= getNodeId()+"-"+id+"-"+procCount++;
        ProcessBuilder proc = new ProcessBuilder(splitCmdLine(commandLine));
        proc.directory( new File("./") );
        try {
            Process p = proc.start();
            ProcessEntry en = new ProcessEntry(commandLine,id,weight,p);
            processes.add(en);
            processEntryOrException.sendResult(en);
        } catch (IOException e) {
            FCLog.log(e);
            processEntryOrException.sendResult(e.getClass().getSimpleName()+"#"+e.getMessage());
        }
    }

    @RemoteMethod(1)
    public void killAllProcesses() {
        System.out.println("killing all processes ..");
        for ( ProcessEntry p : processes ) {
            p.getProc().destroy();
        }
        processes.clear();
    }

    @RemoteMethod(2)
    public void delBaseDir() {
        copyEx.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("deleting base dir");
                try {
                    Files.walkFileTree(FileSystems.getDefault().getPath(outDir), new FileVisitor<Path>() {
                        @Override
                        public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                            return FileVisitResult.CONTINUE;
                        }

                        @Override
                        public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                            Files.delete(file);
                            return FileVisitResult.CONTINUE;
                        }

                        @Override
                        public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
                            return FileVisitResult.CONTINUE;
                        }

                        @Override
                        public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
                            Files.delete(dir);
                            return FileVisitResult.CONTINUE;
                        }
                    });
                } catch (IOException e) {
                    FCLog.log(e);
                }
            }
        });
    }

    Executor copyEx = new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());

    @RemoteMethod(3)
    public void deploy( String relativeDir0, final String filename, final byte[] content, FCFutureResultHandler result ) {
        if ( relativeDir0.startsWith("$LIB") ) {
            outDir = "./lib/update";
            new File(outDir).mkdirs();
            relativeDir0 = relativeDir0.substring("$LIB".length());
        }
        final String relativeDir = relativeDir0;
        copyEx.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    if ( filename.startsWith("local.") ) {
                        String hostName = InetAddress.getLocalHost().getHostName();
                        hostName = hostName.toLowerCase();
                        String prefix = "local." + hostName+".";
                        if ( filename.toLowerCase().startsWith(prefix) ) {
                            FileOutputStream fout = null;
                            try {
                                String targetDir = outDir + "/../local/";
                                String newfilename = filename.substring(prefix.length());
                                if (!new File(targetDir).exists()) {
                                    new File(targetDir).mkdirs();
                                }
                                System.out.println("deploy to local "+filename+" as "+newfilename);
                                fout = new FileOutputStream(targetDir + "/" + newfilename);
                                fout.write(content);
                            } finally {
                                if (fout != null) {
                                    fout.close();
                                }
                            }
                        }
                    } else {
                        System.out.println("deploying " + outDir + relativeDir + "/" + filename);
                        FileOutputStream fout = null;
                        try {
                            String targetDir = outDir + "/" + relativeDir;
                            if (!new File(targetDir).exists()) {
                                new File(targetDir).mkdirs();
                            }
                            fout = new FileOutputStream(targetDir + "/" + filename);
                            fout.write(content);
                        } finally {
                            if (fout != null) {
                                fout.close();
                            }
                        }
                    }
                } catch (IOException e) {
                    FCLog.get().severe("deployment failed ",e);
//                    result.sendResult(e.getClass().getName()+"#"+e.getMessage());
                }
            }
        });
        result.sendResult(true);
    }

    @RemoteMethod(4)
    public void sync(FCFutureResultHandler dummy) {
        while( ((ThreadPoolExecutor)copyEx).getActiveCount() > 0 ){
            Thread.yield();
//            System.out.println(" "+((ThreadPoolExecutor)copyEx).getActiveCount());
        }
        dummy.sendResult("Ok");
    }

    @RemoteMethod(12)
    public void terminate() {
        System.out.println("received termination command ");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.exit(1);
    }


    public static void main( String arg[] ) throws IOException {
        System.out.println("located java in "+findCurrentJavaBin()+".\n (If this is not correct set JAVA_HOME environment)");
        FCDemon.args = arg;
        FCRemoting rem = FastCast.getRemoting();
        if ( arg == null || arg.length != 1) {
            arg = new String[]{null};
        }
        rem.joinCluster("fcdemon.yaml", "FCDemon", arg[0]);
        System.out.println("---------------------------------------------------------------------");
        System.out.println("FastCast cluster demon "+rem.getNodeId());
        System.out.println("---------------------------------------------------------------------");
        System.out.println();
        rem.startReceiving("fcdemon");
        rem.startSending("fcdemon");

    }

}
